package com.study.flink.dataset

import org.apache.commons.io.FileUtils
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration

/**
  * 分布式缓存，（将文件分发到所有节点）
  * @author stephen
  * @create 2019-05-26 23:17
  * @since 1.0.0
  */
object FlinkDistributedCacheDemo {

  def main(args: Array[String]): Unit = {
    // 1、执行环境
    val env = ExecutionEnvironment.getExecutionEnvironment

    // 2、指定数据源
    // step 1：注册一个本地/HDFS文件
    env.registerCachedFile("/Users/stephen/person.txt", "person")
    import org.apache.flink.api.scala._
    val dataStream = env.fromElements("hadoop", "spark", "storm", "flink")

    // 3、对数据进行操作
    val resultStream = dataStream.map(new RichMapFunction[String, String] {

      override def open(parameters: Configuration): Unit = {
        // step 2: 读取到文件
        val personFile = getRuntimeContext.getDistributedCache.getFile("person")
        // 返回一个Java的List
        val lines = FileUtils.readLines(personFile)
        // 转成scala格式的
        import scala.collection.JavaConverters._
        for (line <- lines.asScala) {
          println(line)
        }
      }

      override def map(in: String): String = {
        in
      }
    })

    // 4、指定数据输出位置
    resultStream.print()

  }
}
